package cn._51doit.flink.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MyAtLeastOnceSourceDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(30000);

        DataStreamSource<String> lines1 = env.addSource(new MyAtLeastOnceSource("/Users/start/Desktop/files"));

        DataStreamSource<String> lines2 = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<String> errors = lines2.map(new MapFunction<String, String>() {
            @Override
            public String map(String e) throws Exception {
                if (e.startsWith("error")) {
                    throw new RuntimeException("数据出问题了!!!");
                }
                return e;
            }
        });

        //将两个流合并到一起，是为了任务的制造错误，让Source对应的subtask重启
        DataStream<String> union = lines1.union(errors);
        union.print();


        env.execute();

    }

}
